package ex.tableapi;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public abstract class ApiFrame {
    private String outName = "printOutTable";
    protected TableEnvironment tEnv;

    protected void getEnv() {
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();

        this.tEnv = TableEnvironment.create(settings);
    }

    protected Table registerMysql(String tablename, String flinkTableName) {
        String createOutDDL = "CREATE TABLE " + flinkTableName + " (" +
                "a STRING, " +
                "b BIGINT, " +
                "c STRING, " +
                "d STRING, " +
                "e STRING, " +
                "f STRING, " +
                "rowtime TIMESTAMP(3) ," +
                "rtime TIMESTAMP(3) ," +
                "ltime TIMESTAMP(3) ," +
                "ptime TIMESTAMP(3) ," +
                "WATERMARK FOR rowtime AS rowtime - INTERVAL '5' SECOND" +
                ") WITH (" +
                "   'connector'  = 'jdbc'," +
                "   'url'        = 'jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=Asia/Shanghai&useSSL=false'," +
                "   'table-name' = '" + tablename + "'," +
                "   'driver'     = 'com.mysql.cj.jdbc.Driver'," +
                "   'username'   = 'root'," +
                "   'password'   = 'liji@2024'," +
                "   'sink.buffer-flush.max-rows' = '0' " +
                ")";
        tEnv.executeSql(createOutDDL);
        Table ft = tEnv.from(flinkTableName);
        return ft;
    }
    protected Table registerMysql1(String tablename, String flinkTableName) {
        String createOutDDL = "CREATE TABLE " + flinkTableName + " (" +
                "a STRING, " +
                "b BIGINT, " +
                "c STRING, " +
                "d STRING, " +
                "e STRING, " +
                "f STRING, " +
                "rowtime TIMESTAMP(3) ," +
                "rtime TIMESTAMP(3) ," +
                "ltime TIMESTAMP(3) ," +
                "proctime TIMESTAMP(3) ," +
//                "WATERMARK FOR proctime AS proctime - INTERVAL '5' SECOND" +
                "WATERMARK FOR rowtime AS rowtime - INTERVAL '5' SECOND" +
                ") WITH (" +
                "   'connector'  = 'jdbc'," +
                "   'url'        = 'jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=Asia/Shanghai&useSSL=false'," +
                "   'table-name' = '" + tablename + "'," +
                "   'driver'     = 'com.mysql.cj.jdbc.Driver'," +
                "   'username'   = 'root'," +
                "   'password'   = 'liji@2024'," +
                "   'sink.buffer-flush.max-rows' = '0' " +
                ")";
        tEnv.executeSql(createOutDDL);
        Table ft = tEnv.from(flinkTableName);
        return ft;
    }

    protected Table registerMysql2(String tablename, String flinkTableName) {
        String createOutDDL = "CREATE TABLE " + flinkTableName + " (" +
                "a STRING, " +
                "b BIGINT, " +
                "c STRING, " +
                "d STRING, " +
                "e STRING, " +
                "f STRING, " +
                "rowtime TIMESTAMP(3) ," +
                "rtime TIMESTAMP(3) ," +
                "ltime TIMESTAMP(3) ," +
                "proctime TIMESTAMP(3), "+
                "pt AS PROCTIME() ,"+ // 声明一个额外的列作为处理时间属性
                "WATERMARK FOR rowtime AS rowtime - INTERVAL '5' SECOND" +//声明 rowtime 是事件时间属性，并且用 延迟 5 秒的策略来生成 watermark
                ") WITH (" +
                "   'connector'  = 'jdbc'," +
                "   'url'        = 'jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=Asia/Shanghai&useSSL=false'," +
                "   'table-name' = '" + tablename + "'," +
                "   'driver'     = 'com.mysql.cj.jdbc.Driver'," +
                "   'username'   = 'root'," +
                "   'password'   = 'liji@2024'," +
                "   'sink.buffer-flush.max-rows' = '0' " +
                ")";
        tEnv.executeSql(createOutDDL);
        Table ft = tEnv.from(flinkTableName);
        return ft;
    }

    protected static String createPrintOutDDL(String printSchema) {
        String createPrintOutDDL = "CREATE TABLE printOutTable " + printSchema +
                " WITH (" +
                " 'connector' = 'print' " +
                ")";
        String out = createPrintOutDDL
                .replace("*ROWTIME*", "")
                .replace(",\n" +
                "  WATERMARK FOR `rowtime`: TIMESTAMP(3) AS `rowtime` - INTERVAL '5' SECOND","")
                .replace(" NOT NULL *PROCTIME*","");
        System.out.println("========================createPrintOutDDL==================");
        System.out.println(out);
        System.out.println("========================createPrintOutDDL==================");
        return out;
    }

    protected void registerConsole(String createPrintOutDDL, String outName) {
        tEnv.executeSql(createPrintOutDDL);
    }

}
